from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
import pandas as pd
from bqtool.utils.tools import get_ms_time, to_pdtime_ms


def pd_save_es(df, index_name, type_name, es):
    count = 0
    d = df.fillna('').T.to_dict()
    actions = []
    for v in d.values():
        # 创建action
        action = {
            "_index": index_name,
            "_type": type_name,
            "_source": v
        }
        actions.append(action)
        count += 1
        # 批量处理
        if (len(actions) == 5000):
            try:
                success, _ = bulk(es, actions, raise_on_error=True)
                del actions[:]
                print("第{}条成功导入".format(count))
            except Exception as e:
                print("第{}条导入失败，详细信息查看log日志".format(count))
    if (len(actions) > 0):
        try:
            success, _ = bulk(es, actions, raise_on_error=True)
            print("成功导入")
        except Exception as e:
            print("导入失败，详细信息查看log日志")
    return count


def save_csv():
    df = pd.read_excel('E:\EHL\CODE\python\项目\道安云\测试数据/test数据/卡口过车数据.xlsx')
    df = df[['kkbh', 'gcsj', 'fxlx', 'hphm', 'hpzl', 'cdh']]
    df = df.rename(
        columns={'kkbh': 'bay_id', 'gcsj': 'timestamp', 'fxlx': 'travel_orientation', 'hphm': 'car_plate_number',
                 'hpzl': 'car_plate_type', 'cdh': 'lane_id'})
    df.loc[df.car_plate_number == '-', 'car_plate_number'] = ''
    df.to_csv('E:\EHL\CODE\python\项目\道安云\测试数据/pass.csv', index=False)


if __name__ == '__main__':
    esCli = Elasticsearch(['10.2.111.60'], http_auth=('admin', 'admin'), port=9200)
    f = open(r'E:\EHL\CODE\python\项目\道安云\测试数据/pass.csv',encoding="utf-8")
    df = pd.read_csv(f)
    df.bay_id = df.bay_id.astype(str)
    df.car_plate_type = df.car_plate_type.astype(str)

    df.car_plate_type = df.car_plate_type.str.zfill(2)
    # df = df[(df['timestamp']>'2018-11-01 00:00:00') & (df['timestamp']<'2018-11-03 00:00:00')]
    df.timestamp = df.timestamp.apply(get_ms_time)

    pd_save_es(df,'ehlindex201811', 'pass_car',esCli)

    print('ok')
